package com.nyist.mq.consumer;

import com.alibaba.fastjson.JSONObject;
import com.nyist.entity.User;
import com.nyist.mq.constant.RocketTagConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

/**
 * 1.消费带有生产者事务的信息
 *
 * @ClassName TXConsumerServiceImpl
 * @Description
 * @Author xiaoyang
 * @Date 2023/03/29/16:52
 */
@Slf4j
@Service
@RocketMQMessageListener(
        topic = "NYIST_TOPIC",
        consumerGroup = "consumerGroup" + "_" + RocketTagConstant.TAG_TX_CHARGE,
        selectorExpression = RocketTagConstant.TAG_TX_CHARGE,
        messageModel = MessageModel.BROADCASTING // 广播
)
public class TXConsumerServiceImpl implements RocketMQListener<User> {

    @Override
    public void onMessage(User message) {
        // 一般真实环境这里消费前，得做幂等性判断，防止重复消费
        // 方法一：如果你的业务中有某个字段是唯一的，有标识性，如订单号，那就可以用此字段来判断
        // 方法二：新建一张消费记录表t_mq_consumer_log，字段consumer_key是唯一性，能插入则表明该消息还未消费，往下走，否则停止消费
        // 我个人建议用方法二，根据你的项目业务来定义key，这里我就不做幂等判断了，因为此案例只是模拟，重在分布式事务
        String assertUserId = "123";
        log.info("=====消费者>>>tag=" + RocketTagConstant.TAG_TX_CHARGE + ", 接收到消息为:{}", JSONObject.toJSONString(message));
        // 消费者 处理业务逻辑
        String userId = message.getUserId();
        if (StringUtils.equals(assertUserId, userId)) {
            log.info("======>>>消费者成功消费,userId={}", userId);
        } else {
            log.error("======>>>消费者,消费失败,userId={}", userId);
            throw new RuntimeException("=====>>>error, fail consumer");
        }

        // 给用户增加积分
        // int i = creditMapper.addNumber(userCharge.getUserId(), userCharge.getChargeAmount());
        // if (1 == i) {
        //     log.info("【MQ消费】用户增加积分成功，userCharge={}", JSONObject.toJSONString(userCharge));
        // } else {
        //     log.error("【MQ消费】用户充值增加积分消费失败，userCharge={}", JSONObject.toJSONString(userCharge));
        // }

    }
}
